/**
 * 
 */
package stc.skymobi.fsm;

import java.lang.reflect.InvocationTargetException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.beanutils.PropertyUtils;
import org.apache.commons.collections.Closure;
import org.apache.commons.collections.Predicate;
import org.apache.commons.collections.Transformer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import stc.skymobi.ebus.EventBus;
import stc.skymobi.fsm.event.TimeoutEvent;
import stc.skymobi.fsm.statistic.FSMStatisticer;


/**
 * @author hp
 *
 */
public class DefaultFSM implements FiniteStateMachine {
	
    private static final Logger logger = 
    	LoggerFactory.getLogger(DefaultFSM.class);
    
	private	Map<Object, MutableFSMContext>	contextHolder
		= new ConcurrentHashMap<Object, MutableFSMContext>();
	
	private String				name;
	private	int					fsmId;
	private	ContextKeyFactory	keyFactory;
	private Predicate			initeventAcceptor;
	private	FSMContextFactory	contextFactory;
	private	StateRepository		stateRepository;
	private	ExecutorService[]	execServices = null;
	private	int					executorPoolSize = 1;
    private AtomicInteger   	serviceIdx = new AtomicInteger(0);
    private EventBus			eventBus;
    private String				eventStateChanged;
    private	boolean				logStateChanged = true;
    private	AtomicInteger		ctxCount = new AtomicInteger(0);
    private	boolean				checkBusy = true;
    private	int					busyThreshold = 1000;
    private	Closure				busyHandler = null;
    private	Transformer			snapshotFactory = null;
    private	AtomicInteger 		handledTransactionCount = new AtomicInteger(0);
    private	AtomicInteger 		bypassTransactionCount = new AtomicInteger(0);
    private FSMStatisticer		statisticer;
    private ScheduledExecutorService	timerExecService = null;
    private	Map<TimeoutEvent, Long>		timerEvents = new HashMap<TimeoutEvent, Long>();
    private	long				timerEventCheckInterval = 1000;
    
	public Transformer getSnapshotFactory() {
		return snapshotFactory;
	}

	public void setSnapshotFactory(Transformer snapshotFactory) {
		this.snapshotFactory = snapshotFactory;
	}

	public EventBus getEventBus() {
		return eventBus;
	}

	public void setEventBus(EventBus eventBus) {
		this.eventBus = eventBus;
	}

	public String getEventStateChanged() {
		return eventStateChanged;
	}

	public void setEventStateChanged(String eventStateChanged) {
		this.eventStateChanged = eventStateChanged;
	}

	public String getName() {
		return name;
	}

	public DefaultFSM(String name, int id) {
    	this.name = name;
    	this.fsmId = id;
    }
    
	private ExecutorService createServiceOf(final int idx) {
		return Executors.newFixedThreadPool(1, new ThreadFactory(){

			public Thread newThread(Runnable r) {
				return new Thread(r, name + "-" + idx);
			}});
	}
	
	public void start() {
		execServices = new ExecutorService[executorPoolSize];
		
		for ( int idx = 0; idx < executorPoolSize; idx++) {
			execServices[idx] = createServiceOf(idx);
		}
		timerExecService = new ScheduledThreadPoolExecutor(1, new ThreadFactory(){

			public Thread newThread(Runnable r) {
				return new Thread(r, name + "-timer");
			}});
		scheduleCheckTimeoutEvents();
	}
	
	public void stop() {
		timerExecService.shutdownNow();
		
		for ( ExecutorService service : execServices) {
			service.shutdownNow();
		}
		
		execServices = null;
	}
	
	public int getExecutorPoolSize() {
		return executorPoolSize;
	}

	public void setExecutorPoolSize(int poolSize) {
		this.executorPoolSize = poolSize;
	}

	public String getStateRepositoryAsString() {
		return stateRepository.toString();
	}
	
	public StateRepository getStateRepository() {
		return stateRepository;
	}
	
	public void setStateRepository(StateRepository stateRepository) {
		this.stateRepository = stateRepository;
	}
	
	public FSMContextFactory getContextFactory() {
		return contextFactory;
	}
	
	public String getContextFactoryAsString() {
		return contextFactory.toString();
	}
	
	public void setContextFactory(FSMContextFactory contextFactory) {
		this.contextFactory = contextFactory;
	}
	
	public ContextKeyFactory getKeyFactory() {
		return keyFactory;
	}
	
	public void setKeyFactory(ContextKeyFactory keyFactory) {
		this.keyFactory = keyFactory;
	}
	
	public int getContextCount() {
		return	ctxCount.get();
	}
	
	public String[] getPendingContextDetail() {
		Set<Map.Entry<Object, MutableFSMContext>> ctxs = contextHolder.entrySet();
		
		List<String> ret = new ArrayList<String>();
		SimpleDateFormat formater = new SimpleDateFormat("yyMMdd-HH:mm:ss.SSS");
		
		for ( Map.Entry<Object, MutableFSMContext> entry : ctxs ) {
			StringBuilder sb = new StringBuilder();
			
			sb.append(entry.getKey().toString());
			sb.append(":[");
			
			FSMContext ctx = entry.getValue();
			sb.append(ctx.getCurrentState());
			sb.append("]create:");
			sb.append(formater.format( new Date(ctx.getCreateTime())) );
			sb.append("/last:");
			sb.append(formater.format( new Date(ctx.getLastModify())) );
			ret.add(sb.toString());
		}
		
		return	ret.toArray(new String[0]);
	}
	
	private	int getAndIncServiceIndex() {
        //decide which idx to get session
        return	(serviceIdx.getAndIncrement()) % execServices.length;
	}
	
	private	ExecutorService selectService() {
		return	execServices[getAndIncServiceIndex()];
	}

	private void removeContext(MutableFSMContext ctx) {
		if ( null != this.statisticer ) {
			try {
				statisticer.incHandledTransactionEnd(this.name, this.fsmId, ctx.getEndReason());
			}
			catch (Exception e) {
				logger.error("[" + name + "]incHandledTransactionEnd:", e);
			}
		}
		
		Object key = ctx.getKey();
		
		this.keyFactory.removeKey(key);
		if ( null != this.contextHolder.remove(key) ) {
			//	已做移除操作
			this.ctxCount.decrementAndGet();
		}
		
		try {
			ctx.destroy();
		}
		catch (Exception e) {
			logger.error("[" + name + "]removeContext:", e);
		}
	}

	private State getStateOf(String stateName) {
		return	this.stateRepository.find(stateName);
	}
	
	private void fireStateChanged(FSMContext ctx, String fromState, String toState, Object event) {
		if ( logStateChanged && null != eventBus && null != eventStateChanged ) {
			Object snapshot = ctx.getSnapshot();
			
			if ( null != snapshot ) {
				try {
					PropertyUtils.setProperty(snapshot, "fsm", name);
					PropertyUtils.setProperty(snapshot, "fromState", fromState);
					PropertyUtils.setProperty(snapshot, "toState", toState);
					PropertyUtils.setProperty(snapshot, "eventRecved", 
							null != event ? event.toString() : "");
					PropertyUtils.setProperty(snapshot, "time", new Date());
				} catch (IllegalAccessException e) {
					logger.error("fireStateChanged:", e);
				} catch (InvocationTargetException e) {
					logger.error("fireStateChanged:", e);
				} catch (NoSuchMethodException e) {
					logger.error("fireStateChanged:", e);
				}
				
				eventBus.fireEvent(this.eventStateChanged, snapshot);
			}
		}
	}
	
	private	void initSnapshot(MutableFSMContext ctx) {
		if ( this.logStateChanged && (null != this.snapshotFactory) ) {
			try {
				ctx.setSnapshot(snapshotFactory.transform(ctx));
			}
			catch (Exception e) {
				logger.error("[" + name + "]initSnapshot:", e);
			}
		}
	}
	
	private void initContext(MutableFSMContext ctx, Object event) {
		State state = getStateOf(ctx.getCurrentState());
		if ( null == state ) {
			logger.warn("Can not found valid state for [" + ctx.getCurrentState() 
					+ "] , remove context [" + ctx + "]");
			removeContext(ctx);
		}
		else {
			
			initSnapshot(ctx);
			
			boolean ret = false;
			
			try {
				ret = state.enter(this, ctx);
			}
			catch (Exception e) {
				logger.error( ctx.getCurrentState() + ".enter:", e);
			}
			
			if ( !ret ) {
				if ( logger.isDebugEnabled() ) {
					logger.debug("enter return false, ctx [" + ctx + "] will end normally.");
				}
				removeContext(ctx);
				return;
			}
			fireStateChanged(ctx, "", ctx.getCurrentState(), event);
			
			initSnapshot(ctx);
			
			dispatchEvent(ctx, event);
		}
	}
	
	private void dispatchEvent(MutableFSMContext ctx, Object event) {
		State currentState = getStateOf(ctx.getCurrentState());
		if ( null == currentState ) {
			logger.error("Internal Error: Can not found valid state for [" + ctx.getCurrentState() 
					+ "] , remove context [" + ctx + "]");
			removeContext(ctx);
		}
		
		String nextStateName = null;

		try {
			nextStateName = currentState.acceptEvent(this, ctx, event);
		}
		catch (Exception e) {
			logger.error(ctx.getCurrentState() + ".acceptEvent:", e);
		}
		
		if ( null == nextStateName ) {
			// this context's FSM state reach end
			try {
				currentState.leave(this, ctx);
			}
			catch (Exception e) {
				logger.error(ctx.getCurrentState() + ".leave:", e);
			}
			
			if ( logger.isDebugEnabled() ) {
				logger.debug("ctx [" + ctx + "] will end normally.");
			}
			
			fireStateChanged(ctx, ctx.getCurrentState(), "", event);
			removeContext(ctx);
		}
		else if ( nextStateName.equals(ctx.getCurrentState()) ) {
			// no change
		}
		else {
			State nextState = getStateOf(nextStateName);
			if ( null == nextState ) {
				logger.error("Internal Error: Can not found valid state for [" + nextStateName
						+ "] , just ignore");
			}
			else {
				// next state is valid
				try {
					currentState.leave(this, ctx);
				}
				catch (Exception e) {
					logger.error(ctx.getCurrentState() + ".leave:", e);
				}
				
				String prevStateName = ctx.getCurrentState();
				ctx.setCurrentState(nextStateName);
				
				boolean ret = false;
				try {
					ret = nextState.enter(this, ctx);
				}
				catch (Exception e) {
					logger.error( ctx.getCurrentState() + ".enter:", e);
				}
				
				fireStateChanged(ctx, prevStateName, ctx.getCurrentState(), event);
				initSnapshot(ctx);
				
				if ( !ret ) {
					if ( logger.isDebugEnabled() ) {
						logger.debug("enter return false, ctx [" + ctx + "] will end normally.");
					}
					
					removeContext(ctx);
				}
			}
		}
	}
	
	private void incBypassTransaction() {
		bypassTransactionCount.incrementAndGet();
		if ( null != this.statisticer ) {
			try {
				statisticer.incBypassTransaction(this.name, this.fsmId);
			}
			catch (Exception e) {
				logger.error("[" + name + "]incBypassTransaction:", e);
			}
		}
	}
	
	private void incHandledTransactionStart() {
		handledTransactionCount.incrementAndGet();
		if ( null != this.statisticer ) {
			try {
				statisticer.incHandledTransactionStart(this.name, this.fsmId);
			}
			catch (Exception e) {
				logger.error("[" + name + "]incHandledTransactionStart:", e);
			}
		}
	}
	
	private boolean handleMorethanLimit(Object event) {
		if ( this.checkBusy && this.getContextCount() >= this.busyThreshold ) {
			
			incBypassTransaction();
			
			//	超过CTX限制值
			if ( null != busyHandler ) {
				try {
					busyHandler.execute(event);
				}
				catch (Exception e) {
					logger.error(busyHandler.toString() + ".execute:", e);
				}
			}
			else {
				logger.error("reach ctx limit, and missing BusyHandler.");
			}
			return	true;
		}
		return	false;
	}

	public void acceptEvent(final Object event) {
		Object key = keyFactory.transformToKey(event);
		
		if ( null == key ) {
			logger.warn("event " + event + " can not transform key, just ignore");
			return;
		}
		//	lookup key in contextHolder
		final MutableFSMContext ctx = contextHolder.get(key);
		if ( null == ctx ) {
			if ( !this.initeventAcceptor.evaluate(event)) {
				if ( logger.isDebugEnabled() ) {
					logger.debug("event [" + event + "] not acceptable, just ignore");
				}
				return;
			}
			
			if ( handleMorethanLimit(event) ) {
				if ( logger.isDebugEnabled() ) {
					logger.debug("FSM[" + name + "] busy, event[" + event + "] has been dropped.");
				}
				return;
			}
			
			final MutableFSMContext initCtx = contextFactory.createFSMContext(this, event);
			if ( null == initCtx ) {
				logger.warn("Can not create context for event " + event + " , just ignore");
				return;
			}
			
			incHandledTransactionStart();
			
			ExecutorService service = this.selectService();
			
			if ( null == service ) {
				throw new RuntimeException("Internal Error: can not select a valid executor service.");
			}
			
			//	init ctx, and continue
			initCtx.setService(service);
			initCtx.setKey(key);
			
			if ( null == contextHolder.put(key, initCtx) ) {
				// add new 
				ctxCount.incrementAndGet();
			}
			
			service.submit(new Runnable(){

				public void run() {
					initContext(initCtx, event);
				}});
		}
		else {
			ctx.getService().submit(new Runnable(){

				public void run() {
					dispatchEvent(ctx, event);
				}});
		}
	}
	
	//	in timerExecService thread
	private void doScheduleTimeout(TimeoutEvent event, long delay) {
		timerEvents.put(event, System.currentTimeMillis() + delay);
	}
	
	//	in timerExecService thread
	private void doCancelTimeout(TimeoutEvent event) {
		if ( null == timerEvents.remove(event) ) {
			if ( logger.isWarnEnabled() ) {
				logger.warn("Can not found matched timeout event[" + event + "], maybe has been emited.");
			}
		}
	}
	
	private void scheduleCheckTimeoutEvents() {
		this.timerExecService.schedule(new Runnable(){

			public void run() {
				doCheckTimeoutEvents();
			}}, 
			this.timerEventCheckInterval, TimeUnit.MILLISECONDS);
	}
	
	//	in timerExecService thread
	private void doCheckTimeoutEvents() {
		long now = System.currentTimeMillis();
		
		try {
			Set<Map.Entry<TimeoutEvent, Long>> entrySet = timerEvents.entrySet();
			
			Iterator<Map.Entry<TimeoutEvent, Long>> iter = entrySet.iterator();
			while ( iter.hasNext() ) {
				Map.Entry<TimeoutEvent, Long> entry = iter.next();
				if ( entry.getValue() <= now ) {
					//	condition matched, emit timeout event
					iter.remove();
					this.acceptEvent(entry.getKey());
				}
			}
		}
		catch (Exception e) {
			logger.error("doCheckTimeoutEvents:", e);
		}
		finally {
			scheduleCheckTimeoutEvents();
		}
	}
	
	public Runnable scheduleTimeout(final TimeoutEvent event, final long delay) {
		this.timerExecService.submit(new Runnable(){

			public void run() {
				doScheduleTimeout(event, delay);
			}});
		
		return new Runnable(){

			public void run() {
				timerExecService.submit(new Runnable(){

					public void run() {
						doCancelTimeout(event);
					}});
			}};
	}

	private int getPendingTaskCountOf(ExecutorService exec) {
    	if ( exec instanceof ThreadPoolExecutor ) {
    		BlockingQueue<Runnable> queue = ((ThreadPoolExecutor)exec).getQueue();
    		return queue.size();
    	}
    	else {
    		throw new RuntimeException("Internal Error-EventReactor : exec is !NOT! ScheduledThreadPoolExecutor class");
    	}
	}
	
    public int[] getPendingTaskCount() {
    	int[] ret = new int[this.execServices.length];
    	int idx = 0;
    	for (ExecutorService exec : this.execServices) {
    		ret[idx++] = getPendingTaskCountOf(exec);
    	}
    	
    	return ret;
    }

	/**
	 * @return the initeventAcceptor
	 */
	public String getIniteventAcceptorAsString() {
		return initeventAcceptor.toString();
	}

	public Predicate getIniteventAcceptor() {
		return initeventAcceptor;
	}
	
	/**
	 * @param initeventAcceptor the initeventAcceptor to set
	 */
	public void setIniteventAcceptor(Predicate initeventAcceptor) {
		this.initeventAcceptor = initeventAcceptor;
	}

	public Closure getBusyHandler() {
		return busyHandler;
	}

	public void setBusyHandler(Closure busyHandler) {
		this.busyHandler = busyHandler;
	}

	public boolean isLogStateChanged() {
		return logStateChanged;
	}

	public void setLogStateChanged(boolean logStateChanged) {
		this.logStateChanged = logStateChanged;
	}

	public boolean isCheckBusy() {
		return checkBusy;
	}

	public void setCheckBusy(boolean checkBusy) {
		this.checkBusy = checkBusy;
	}

	public int getBusyThreshold() {
		return busyThreshold;
	}

	public void setBusyThreshold(int busyThreshold) {
		this.busyThreshold = busyThreshold;
	}

	/**
	 * @return the handledTransactionCount
	 */
	public int getHandledTransactionCount() {
		return handledTransactionCount.get();
	}

	/**
	 * @return the bypassTransactionCount
	 */
	public int getBypassTransactionCount() {
		return bypassTransactionCount.get();
	}

	/**
	 * @return the statisticer
	 */
	public FSMStatisticer getStatisticer() {
		return statisticer;
	}

	/**
	 * @param statisticer the statisticer to set
	 */
	public void setStatisticer(FSMStatisticer statisticer) {
		this.statisticer = statisticer;
	}

	/**
	 * @return the fsmId
	 */
	public int getFsmId() {
		return fsmId;
	}

    /**
	 * @return the timerEventCheckInterval
	 */
	public long getTimerEventCheckInterval() {
		return timerEventCheckInterval;
	}

	/**
	 * @param timerEventCheckInterval the timerEventCheckInterval to set
	 */
	public void setTimerEventCheckInterval(long timerEventCheckInterval) {
		this.timerEventCheckInterval = timerEventCheckInterval;
	}

	public int getPendingTimeoutEvent() {
		return	this.timerEvents.size();
	}
	
	public int getPendingTimerTaskCount() {
    	if ( this.timerExecService instanceof ScheduledThreadPoolExecutor ) {
    		BlockingQueue<Runnable> queue = ((ScheduledThreadPoolExecutor)timerExecService).getQueue();
    		return queue.size();
    	}
    	else {
    		throw new RuntimeException("Internal Error-EventReactor : timerExecService is !NOT! ScheduledThreadPoolExecutor class");
    	}
	}

	/* (non-Javadoc)
	 * @see java.lang.Object#toString()
	 */
	@Override
	public String toString() {
		return this.name;
	}
}
